package com.wulaobo.stream;

import com.wulaobo.constant.Constants;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author wuwenbo
 * @Date 2024/5/24 10:06
 * @ClassName: SplitStreamByFilter
 * @Description: TODO
 */
public class SplitStreamByFilter {

    public static void main(String[] args){


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<Integer> ds = env.socketTextStream(Constants.IP, Constants.SOCKET_PORT)
                .map(Integer::valueOf);
        //将ds 分为两个流 ，一个是奇数流，一个是偶数流
        //使用filter 过滤两次

        SingleOutputStreamOperator<Integer> ds1 = ds.filter(x -> x % 2 == 0);
        SingleOutputStreamOperator<Integer> ds2 = ds.filter(x -> x % 2 == 1);

        ds1.print("偶数");
        ds2.print("奇数");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
